<?php
namespace Snitso\IO\Reactor\Swoole\MultiAdvanceTask;

trait TaskTrait
{
    public $onTask = null;
    public $onFinish = null;

    /**
     * @var array task pid 存储数据
     */
    protected $taskPidArr = [];

    /**
     * @var null Message queue
     */
    protected $msgQueue = null;

    public function forkTask()
    {
        $this->msgQueue = $this->msgQueue ?: msg_get_queue(ftok(__FILE__, 'u'));

        for ($i = 0; $i < $this->taskNum; $i++) {
            $taskPid = pcntl_fork();

            if ($taskPid === -1) {
                exit('fork task 子进程错误' . PHP_EOL);
            } elseif ($taskPid === 0) {
                $this->msgReceive();
                exit;
            } else {
                // 父进程空间
                $this->taskPidArr[] = $taskPid;
            }
        }
    }

    protected function msgReceive()
    {
        $pid = posix_getpid();

        while (1) {
            msg_receive($this->msgQueue, $pid, $msgType, 1024, $message);
            ($this->onTask)($this, $message);
        }
    }

    /**
     * Note: 将数据投递给指定 task 进程或随机
     *
     * @param $data
     * @param int $taskPid
     * @author: Zhao Zhiqiang
     * @since: 2021/5/7 10:12
     */
    public function task($data, int $taskPid = 0)
    {
        $msgType = $taskPid ?: $this->taskPidArr[array_rand($this->taskPidArr, 1)];
        msg_send($this->msgQueue, $msgType, $data);
    }

}